package producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.awt.image.renderable.RenderableImageProducer;
import java.io.Serializable;
import java.util.Properties;

/**
 * @Author:RenPu
 * @Date: 2020/3/18 18:42
 * @Version: 1.0
 * @description:生产者
 */
public class KafkaProdecerDemo {

    public static void main(String[] args) {

        //创建配置对象，指定producer的配置
        Properties properties = new Properties();

        //配置与kafka集群连接的参数
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092");

        //配置producer发布的recoder的key进行序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);

        //配置producer发布的recoder的value进行序列化,StringSerializer序列化器
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);


        //创建producer对象
        KafkaProducer<Integer, String> producer = new KafkaProducer<Integer, String>(properties);

        //发布消息
        ProducerRecord<Integer, String> record = new ProducerRecord<Integer, String>("baizhi",1,"nihaowodeshijie");
        producer.send(record);

        //提交
        producer.flush();

       //释放资源
       producer.close();



    }




}
